/*******************************************************************************
 * Copyright (c) 2008 Sylvain Sicard.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the GNU Lesser Public License v2.1
 * which accompanies this distribution, and is available at
 * http://www.gnu.org/licenses/old-licenses/gpl-2.0.html
 * 
 * Contributors:
 *     Sylvain Sicard - initial API and implementation
 ******************************************************************************/
package fr.jade.fraclite.orb.protocols;

import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.Vector;

import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.JChannelFactory;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
import org.jgroups.util.Promise;

import fr.jade.fraclite.orb.api.Marshaller;
import fr.jade.fraclite.orb.io.BasicMarshaller;
import fr.jade.fraclite.orb.io.messages.AdminMessage;
import fr.jade.fraclite.orb.io.messages.MethodCall;
import fr.jade.fraclite.orb.io.messages.MethodReply;
import fr.jade.fraclite.orb.io.messages.MulticastRpcState;

/**
 * @author Sylvain Sicard
 */
public class MulticastRpcServer extends ReceiverAdapter{
  protected static JChannelFactory factoryStub = new JChannelFactory();
    protected Object serverObj;
    protected Channel channel;
    protected Cache cache;
    protected Marshaller rawMarshaller;
    protected Marshaller marshaller;
    Address local_addr=null;
    Vector<Address> members=new Vector<Address>();
    Promise<Boolean> state_promise=new Promise<Boolean>();
    
    public MulticastRpcServer(String groupId, Marshaller marshaller, Object serverObj){
      this.cache = new Cache();
      this.serverObj=serverObj;
      this.marshaller=marshaller;
      this.rawMarshaller = new BasicMarshaller();
      try {
        factoryStub.setMultiplexerConfig("ha-stacks.xml");
        //factoryStub.setMultiplexerConfig("udp.xml");
        //channel = factoryStub.createChannel("udp");
        //channel = factoryStub.createMultiplexerChannel("udp", groupId);
        channel = factoryStub.createMultiplexerChannel("udp", groupId);
        channel.setOpt(Channel.LOCAL, false);
        channel.setReceiver(this);
        channel.connect(groupId);
        
        if(!isCoordinator()){
          //initState();
        }
      
      } catch (Exception e1) {
        e1.printStackTrace();
      }
    }
    
    public boolean isCoordinator() {
      Object first_mbr=null;

      synchronized(members) {
          first_mbr=!members.isEmpty()? members.elementAt(0) : null;
          if(first_mbr == null)
              return true;
      }
      return getLocalAddress() != null && getLocalAddress().equals(first_mbr);
  }
    public Address getLocalAddress() {
      if(local_addr != null) return local_addr;
      if(channel != null)
          local_addr=channel.getLocalAddress();
      return local_addr;
  }
    /* -------------------- Interface MessageListener ------------------- */
    
    @Override
    public byte[] getState(){
      System.out.println("[skel] getState");
      Map<Long, Long> callCounters = new TreeMap<Long, Long>();
      MulticastRpcState state = new MulticastRpcState(callCounters);
      for(String s : cache.keySet()){
        String[] entry = s.split("@");
        Long stubId = new Long(entry[0]);
        Long callId = new Long(entry[1]);
        long max;
        if(callCounters.containsKey(stubId)){
          max = Math.max(callId, callCounters.get(stubId));
        }else{
          max = callId;
        }
        
        callCounters.put(stubId, max);
      }
      System.out.println("Skel state - "+callCounters.toString());
      try {
        return rawMarshaller.objectToByteBuffer(state);
      } catch (Exception e) {
        e.printStackTrace();
        return null;
      }
    }

    @Override
    public void setState(byte[] state){
      System.out.println("[skel] setState");
      try {
        MulticastRpcState s = (MulticastRpcState)rawMarshaller.objectFromByteBuffer(state);
        // TODO
      } catch (Exception e) {
        e.printStackTrace();
      }
      state_promise.setResult(Boolean.TRUE);
    }
    
    @Override
    public void receive(Message msg) {
      try {
        //System.out.println("Receive a message");
        Object o = rawMarshaller.objectFromByteBuffer(msg.getBuffer());
        if (o instanceof MethodCall){
          MethodCall call = (MethodCall)o;
          //System.out.println("receive " + call.toString());
          Object result = handleMethodCall(call.getCallId(), 
                                           call.getMethodName(),
                                           call.getTypes(),
                                           call.getArgs());
          sendResult(call.getCallId(), result);
        }else if (o instanceof AdminMessage){
          if(isCoordinator()){
          Message msgRes = new Message(null, null, getState());
          channel.send(msgRes);
          }
        } else {
          //ignore
          //System.out.println("skel ignore message");
        }
      } catch (Exception e1) {
        e1.printStackTrace();
      }
    }
       
    public synchronized void viewAccepted(View new_view) {
      if(new_view == null) return;
      members=new_view.getMembers();
      }

    /*--------------------------------------------------------------*/
    
    public void sendResult(String callId, Object result){
      try{
      MethodReply reply = new MethodReply(callId, marshaller.objectToByteBuffer(result));
      byte[] buf = marshaller.objectToByteBuffer(reply);
      Message msgRes = new Message(null, null, buf);
      channel.send(msgRes);
      }catch(Exception e){
        e.printStackTrace();
      }
    }
    
    public Object handleGetState() {
      byte[] ret = null;
      synchronized(cache){
        try {
          ret = rawMarshaller.objectToByteBuffer(cache);
        } catch (Exception e) {
          e.printStackTrace();
        }
        return ret;
      }
        
    }
    
    public Object handleMethodCall(String callId, String methodName, Class<?>[] methodTypes, byte[] rawArgs){
      CacheEntry        call = null;
      Object      result = null;
      
      synchronized (cache) {
        //System.out.println(cache.toString());
        if(cache.containsKey(callId)){
          //System.out.println("Duplicate call("+callId+") "+ methodName );
          call = cache.get(callId);
        }else{
          //System.out.println("New call("+callId+") "+ methodName );
          call = cache.put(callId);
        }
      }
      
      synchronized (call) {
        if(call.valid){
          result = call.result;
        }else{
          try{
            Object[] args = (Object[])marshaller.objectFromByteBuffer(rawArgs);
            result = invoke(methodName, methodTypes, args, serverObj);
            call.result = result;
            call.valid = true;
          }catch(Exception e){
            e.printStackTrace();
          } catch (Throwable e) {
            e.printStackTrace();
          }
        }
      }
      //System.out.println("return: "+result);
      return result;
    }
    
    
    /**
     * Invokes the method with the supplied arguments against the target object. If a method lookup is provided, it will be used. Otherwise, the default method
     * lookup will be used.
     * 
     * @param target
     *          - the object that you want to invoke the method on
     * @return an object
     */
    protected Object invoke(String method_name, 
                            Class<?>[] types, 
                            Object[] args, 
                            Object target) throws Throwable {
      Method meth = null;
      Object retval = null;
      Class<?> cl = target.getClass();

      try {
        meth = cl.getMethod(method_name, types);
        retval = meth.invoke(target, args);
      } catch (InvocationTargetException inv_ex) { 
        retval=inv_ex;
      } catch (NoSuchMethodException no) {
        StringBuffer sb = new StringBuffer();
        sb.append("found no method called ").append(method_name).append(" in class ");
        sb.append(cl.getName()).append(" with (");
//        if (args != null) {
//          for (Class<?> c : types) {
//            sb.append(c.getName() + ", ");
//          }
//        }
        sb.append(") formal parameters");
        System.out.println(sb.toString());
        throw no;
      } catch (Throwable e) {
        // e.printStackTrace(System.err);
        System.out.println("exception in invoke()" + e);
        throw e;
      }
      return retval;
    }

    protected class CacheEntry implements Serializable{
      
      private static final long serialVersionUID = 1L;

      public boolean valid;
      
      public Object result;
      
      public CacheEntry(boolean valid){
        this.valid = valid;
      }
    }
    
    protected class Cache implements Serializable{
      private static final long serialVersionUID = 1L;
      protected Map<String, CacheEntry> calls;
      protected List<CacheEntry> callsSequence;
      protected Map<Long, Long> currentId;
      public Cache(){
        this.calls = new HashMap<String, CacheEntry>();
        this.currentId = new TreeMap<Long, Long>();
        this.callsSequence = new LinkedList<CacheEntry>();
      }

      public CacheEntry put(String callId) {
        CacheEntry call = new CacheEntry(false);
        calls.put(callId, call);
        callsSequence.add(call);
        return call;
      }

      public CacheEntry get(String callId) {
        return calls.get(callId);
      }

      public boolean containsKey(String callId) {
        return calls.containsKey(callId);
      }
      
      public Set<String> keySet() {
        return calls.keySet();
      }
      
      public String toString(){
        return calls.keySet().toString();
      }
    }
}
